package com.dwm;

import com.alibaba.fastjson.JSONObject;
import com.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.*;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * 在 DWM 层中，使用 CEP 筛选出跳出的数据明细，并在输出流中发送到相应的 Kafka 主题中（8分）
 * 利用log读
 */
public class Six {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //从kafka中读取log日志数据
        FlinkKafkaConsumer<String> practice = KafkaUtils.createConsumer("ods_base_log","ojh");
        DataStreamSource<String> stream = env.addSource(practice);
        //对stream进行筛选数据  不能转为json的数据发送到侧流
        OutputTag<String> outputTag = new OutputTag<String>("wrong") {};
       // OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>("wrong") {};

        SingleOutputStreamOperator<JSONObject> pageDs = stream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String str, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jo = JSONObject.parseObject(str);
                    if (jo.containsKey("page")) {
                        collector.collect(jo);
                    }
                } catch (Exception e) {
                    context.output(outputTag, str);
                }
            }
        });
        pageDs.print("page:");
        DataStream<String> wrongDs = pageDs.getSideOutput(outputTag);
        //cep进行跳出明细数据
        //1.数据
        //2.规则
        SingleOutputStreamOperator<JSONObject> tsStream = pageDs.keyBy(x -> {
            return x.getJSONObject("common").getString("mid");
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long l) {
                        return jsonObject.getLong("ts");
                    }
                }));


        Pattern<JSONObject, JSONObject> jumpDsPattern = Pattern.<JSONObject>begin("one").where(
                new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject event) {
                        return event.getJSONObject("page").getString("last_page_id") == null;
                    }
                }
        ).next("two").where(
                new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject event) {
                        String page_id = event.getJSONObject("page").getString("page_id");
                        if(page_id!=null || page_id.length()>0){
                            return true;
                        }
                       return false;
                    }
                }

        ).within(Time.milliseconds(500));

        //3.
        PatternStream<JSONObject> ps = CEP.pattern(tsStream, jumpDsPattern);
        //跳出的侧流
        OutputTag<String> jumpTag = new OutputTag<String>("jump") {};
        SingleOutputStreamOperator<String> jumpStream = ps.flatSelect(jumpTag, new PatternFlatTimeoutFunction<JSONObject, String>() {
            @Override
            public void timeout(Map<String, List<JSONObject>> map, long l, Collector<String> collector) throws Exception {
                List<JSONObject> one = map.get("one");
                for (JSONObject jsonObject : one) {
                    collector.collect(jsonObject.toString());
                }
            }
        }, new PatternFlatSelectFunction<JSONObject, String>() {
            @Override
            public void flatSelect(Map<String, List<JSONObject>> map, Collector<String> collector) throws Exception {
                //不超时不提取
            }
        });
        DataStream<String> jump = jumpStream.getSideOutput(jumpTag);
        jump.print();
        jump.addSink(KafkaUtils.createProduer("dwm_jump_log"));

        env.execute();

    }
}
